package com.ibm.cps.processors;

import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Modifier;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.log4j.Logger;
import org.reflections.Reflections;

import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.newmessage.AbstractMetadata;
import com.ibm.cps.newmessage.BasicMetadata;
import com.ibm.cps.newmessage.JointMetadata;
import com.ibm.cps.newmessage.MetadataFactory.MetadataType;
import com.ibm.cps.newmessage.OptimizedMetadata;
import com.ibm.cps.sql.expression.function.meta.FunctionMetaCreator;
import com.ibm.factories.TimeSeriesSchemaFactory;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.Pair;
import com.ibm.util.exception.CPSException;

public class ProcessorFactory {

	private static HashMap<String, Constructor<? extends AbstractProcessor>> type2Constructor = new HashMap<String, Constructor<? extends AbstractProcessor>>();
	private static Logger logger;
	private static ProcessorFactory factory = null;

	static {
		try {
			logger = Logger.getLogger(ProcessorFactory.class);
			// may change to a localconfig setting
			Reflections reflections = new Reflections("com.ibm");
			Set<Class<? extends AbstractProcessor>> processors = reflections.getSubTypesOf(AbstractProcessor.class);
			for(Class<? extends AbstractProcessor> processor : processors){
				// skip the abstract AbstractProcessor's sub class.
				if(Modifier.isAbstract(processor.getModifiers())){
					continue;
				}
				registerProcessor(processor);
			}
//			type2Constructor.put(MessageFields.SIMPLE_INPUT_TYPE , getOneSchemaConstructor(SimpleInputProcessor.class));
//			type2Constructor.put(MessageFields.BATCH_DFT_TYPE, getOneSchemaConstructor(BasicDFTProcessor.class));
//			type2Constructor.put(MessageFields.RULE_TYPE, getOneSchemaConstructor(RuleEngineProcessor.class));
//			type2Constructor.put(MessageFields.PUBLISH_TYPE, getOneSchemaConstructor(OutputProcessor.class));
//			type2Constructor.put(MessageFields.NAIVE_RULE, getOneSchemaConstructor(NaiveRuleEngineProcessor.class));
//			type2Constructor.put(MessageFields.SLIDING_WINDOW_TYPE, getOneSchemaConstructor(WindowProcessor.class));
//			type2Constructor.put(MessageFields.COUNT_WINDOW_TYPE, getOneSchemaConstructor(CountWindowProcessor.class));
//			type2Constructor.put(MessageFields.AVERAGE_TYPE, getOneSchemaConstructor(AverageProcessor.class));
//			type2Constructor.put(MessageFields.VARIANCE_TYPE, getOneSchemaConstructor(VarianceProcessor.class));
//			type2Constructor.put(MessageFields.SCALAR_CALCULATION,
//					getOneSchemaConstructor(ScalarCalculationProcessor.class));
//			type2Constructor.put(MessageFields.VECTOR_CALCULATION,
//					getOneSchemaConstructor(VectorCalculationProcessor.class));
//			type2Constructor.put(MessageFields.VECTOR_PROJECTION,
//					getOneSchemaConstructor(VectorProjectionProcessor.class));
//			//add by gxj
//			type2Constructor.put(MessageFields.DTW_TYPE, getOneSchemaConstructor(DTWProcessor.class));
//			type2Constructor.put(MessageFields.DFT_VECTOR_TYPE, getOneSchemaConstructor(DFTProcessor.class));
//			type2Constructor.put(MessageFields.MAX_TYPE, getOneSchemaConstructor(MaxProcessor.class));
//			type2Constructor.put(MessageFields.MIN_TYPE, getOneSchemaConstructor(MinProcessor.class));
//
//			// add for static join
//			type2Constructor.put(MessageFields.STATIC_JOINT, getOneSchemaConstructor(StaticJointProcessor.class));
//
//			type2Constructor.put(MessageFields.INPUT_TYPE, getOneSchemaConstructor(FullInputProcessor.class));
			
			
			
		} catch (CPSException e) {
			logger.error(e.getMessage());
		}
	}

	private ProcessorFactory() throws CPSException {

	}

	public Map<String, MetadataType> getProcessorType() throws CPSException {
		Map<String, MetadataType> metadata2Type = new HashMap<String, MetadataType>();
		for (Map.Entry<String, Constructor<? extends AbstractProcessor>> entry : type2Constructor.entrySet()) {
			String type = entry.getKey();
			String superClassName = entry.getValue().getDeclaringClass().getSuperclass().getName();

			if (superClassName.equals(JointProcessor.class.getName())) {
				metadata2Type.put(type, MetadataType.JointProcessorMetadata);
			} else if (superClassName.equals(BasicProcessor.class.getName()) || superClassName.equals(SchemaInheritiveProcessor.class.getName())) {
				metadata2Type.put(type, MetadataType.BasicProcessorMetadata);
			} else if (superClassName.equals(OptimizedProcessor.class.getName())) {
				metadata2Type.put(type, MetadataType.OptimizedProcessorMetadata);
			} else {
				throw new CPSException(ErrorCode.INVALID_INPUT_ERROR,
						"Processor should implement one of the following type basic, optimized or joint.");
			}
		}
		return metadata2Type;
	}

	public static ProcessorFactory getInstance() throws CPSException {
		if (factory == null) {
			factory = new ProcessorFactory();
		}
		return factory;
	}

	// TODO
	public static void registerProcessor(Class<? extends AbstractProcessor> processor) throws CPSException {
		if(processor.isAnnotationPresent(Info.class)){
			Info info = processor.getAnnotation(Info.class);
			if(type2Constructor.containsKey(info.Name())){
				String message = info.Name() + " already exists, error happened in: " + processor.getName();
				logger.warn(message);
				throw new CPSException(ErrorCode.ALREADY_EXIST_CODE, message);
			}
			type2Constructor.put(info.Name(),getOneSchemaConstructor(processor));
		}else{
			logger.warn(processor.getName() + " don't hava a Info annotation, please check.");
		}
	}

	@SuppressWarnings("rawtypes")
	public static Constructor<? extends AbstractProcessor> getOneSchemaConstructor(
			Class<? extends AbstractProcessor> cls) throws CPSException {
		Class[] argTypes;
		String superClassName = cls.getSuperclass().getName();
		if (superClassName.equals(JointProcessor.class.getName())) {
//			argTypes = new Class[] { ITimeSeriesSchema.class, ITimeSeriesSchema.class, AbstractMetadata.class };
			argTypes = new Class[] { String.class, String.class, String.class, ITimeSeriesSchema.class, ITimeSeriesSchema.class };
		} else if (superClassName.equals(BasicProcessor.class.getName())
				|| superClassName.equals(OptimizedProcessor.class.getName()) 
				|| superClassName.equals(SchemaInheritiveProcessor.class.getName())) {
//			argTypes = new Class[] { ITimeSeriesSchema.class, AbstractMetadata.class };
			argTypes = new Class[] { String.class, String.class, String.class, ITimeSeriesSchema.class };
		} else {
			throw new CPSException(ErrorCode.INVALID_INPUT_ERROR,
					"Processor should implement one of the following type basic, optimized or joint.");
		}
		try {
			return cls.getConstructor(argTypes);
		} catch (NoSuchMethodException | SecurityException e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, Throwables.getStackTraceAsString(e));
		}
	}

	public static AbstractProcessor createProcessor(String type) throws CPSException {
		Constructor<? extends AbstractProcessor> cons = type2Constructor.get(type);
		if (cons == null) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "There is no constructor for type " + type);
		}

		try {
			return (AbstractProcessor) cons.newInstance();
		} catch (InstantiationException | IllegalAccessException | IllegalArgumentException
				| InvocationTargetException e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "Fail to construct processor: " + e.getCause());
		}
	}

	public AbstractProcessor createProcessor(AbstractMetadata metadata, HashMap<String, AbstractProcessor> id2Processor)
			throws CPSException {
		if (id2Processor.keySet().contains(metadata.getProcessorid())) {
			logger.info("Processor " + metadata.getProcessorid() + " was existed.");
			return null;
		}
		Constructor<? extends AbstractProcessor> cons = type2Constructor.get(metadata.getInputFieldType());
		if (cons == null) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"There is no constructor for type " + metadata.getInputFieldType());
		}
		Preconditions.checkNotNull(cons, "There is no constructor for type " + metadata.getInputFieldType());
		try {
			if (metadata instanceof BasicMetadata || metadata instanceof OptimizedMetadata) {
				ITimeSeriesSchema parentSchema = getParentSchema(metadata, id2Processor);
				Preconditions.checkNotNull(parentSchema, "Null parent schema.");
//				Object[] args = new Object[] { parentSchema, metadata };
				Object[] args = new Object[] { metadata.getTenantid(), metadata.getProcessorid(), metadata.getSource(), parentSchema };
				AbstractProcessor processor = (AbstractProcessor) cons.newInstance(args);
				FunctionMetaCreator creator = new FunctionMetaCreator(processor.getClass());
				creator.assignProcessorAttributes(processor, metadata.getKey2value());
				processor.open();
				return processor;
			} else if (metadata instanceof JointMetadata) {
				Pair<ITimeSeriesSchema, ITimeSeriesSchema> parentSchemas = getParentSchemas(metadata, id2Processor);
				Preconditions.checkNotNull(parentSchemas, "Null parent schema.");
//				Object[] args = new Object[] { parentSchemas.getFirst(), parentSchemas.getSecond(), metadata };
				Object[] args = new Object[] { metadata.getTenantid(), metadata.getProcessorid(), metadata.getSource(), parentSchemas.getFirst(), parentSchemas.getSecond()};
				AbstractProcessor processor = (AbstractProcessor) cons.newInstance(args);
				FunctionMetaCreator creator = new FunctionMetaCreator(processor.getClass());
				creator.assignProcessorAttributes(processor, metadata.getKey2value());
				processor.open();
				return processor;
			}
		} catch (InvocationTargetException e) {
			Throwable exception = e.getTargetException();
			if (exception instanceof CPSException) {
				throw (CPSException) exception;
			}
			e.printStackTrace();
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "Fail to construct processor: " + e.getCause());
		} catch (CPSException e) {
			throw e;
		} catch (Exception other) {
			other.printStackTrace();
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "Internal error.");
		}
		return null;
	}
	
	public void updateProcessorMap(String metadataid, AbstractProcessor processor,
			HashMap<String, AbstractProcessor> id2Processor, AbstractMetadata processorMeta) throws CPSException {

		if (processor instanceof BasicProcessor) {
			String processorid = processor.getProcessorId();
			id2Processor.put(processorid, processor);
			logger.info("Add processor " + processorid + ", type is " + processorMeta.getInputFieldType());
		} else if (processor instanceof OptimizedProcessor) {
			List<String> parents = processorMeta.getParentProcessors();
			for (String parentId : parents){
				String connectorIdToParent = parentId + processorMeta.getInputFieldType();
				if (id2Processor.get(connectorIdToParent) == null) {
					id2Processor.put(connectorIdToParent, processor);
					logger.info("Created connector rule node " + connectorIdToParent);
				}
				OptimizedProcessor connector = (OptimizedProcessor) id2Processor.get(connectorIdToParent);
				connector.addSubProcessor(processorMeta);
			}
			id2Processor.put(metadataid, processor);
			logger.info("Add rule to rule engine " + metadataid);
		} else {
			// Joint metadata
		}
	}

	private Pair<ITimeSeriesSchema, ITimeSeriesSchema> getParentSchemas(AbstractMetadata metadata,
			HashMap<String, AbstractProcessor> id2Processor) {
		return null;
	}

	private ITimeSeriesSchema getParentSchema(AbstractMetadata metadata,
			HashMap<String, AbstractProcessor> id2Processor) throws CPSException {
		if (metadata instanceof BasicMetadata || metadata instanceof OptimizedMetadata) {

			// input metadata
			if (ProcessorFactory.isInput(metadata.getInputFieldType())) {
				ITimeSeriesSchema schema = TimeSeriesSchemaFactory.createFromOtherMetadata(metadata.getTenantid(),
						metadata.getKey2value().get(MessageFields.SCHEMAID),
						JSONUtility.fromObjectString(metadata.getInput()));
				if (schema == null) {
					throw new CPSException(ErrorCode.KEY_NOT_EXIST_CODE,
							"Schema " + metadata.getKey2value().get(MessageFields.SCHEMAID) + " didn't exist.");
				}
				return schema;
			} else {
				ITimeSeriesSchema schema = null;
				for (String parentProcessor : metadata.getParentProcessors()) {
					if (id2Processor.get(parentProcessor) == null
							|| id2Processor.get(parentProcessor).getTimeSeriesSchema() == null) {
						throw new CPSException(ErrorCode.KEY_NOT_EXIST_CODE,
								"Parent processor " + parentProcessor + " didn't exist.");
					}
					if (schema == null) {
						schema = id2Processor.get(parentProcessor).getTimeSeriesSchema();
					} else if (schema.compareTo(id2Processor.get(parentProcessor).getTimeSeriesSchema()) != 0) {
						String info = "Parent schemas for basic processor should be the same.\n";
						info += "One schema is " + schema.getSchemaID() + ": " + schema.getTags();
						info += "Another schema is "
								+ id2Processor.get(parentProcessor).getTimeSeriesSchema().getSchemaID() + ": "
								+ id2Processor.get(parentProcessor).getTimeSeriesSchema().getTags();
						throw new CPSException(ErrorCode.INPUT_ERROR, info);
					}
				}
				return schema;
			}
		} else {
			return null;
		}
	}

	public void deleteProcessorMap(HashMap<String, AbstractProcessor> id2Processor, AbstractProcessor processor,
			String source) throws CPSException {
		if (processor instanceof BasicProcessor) {
			id2Processor.remove(processor.getProcessorId());
			processor.dispose();
			logger.info("Basic Processor " + processor.getProcessorId() + " has been deleted.");
		} else if (processor instanceof OptimizedProcessor) {
			deleteOptimizedProcessor(id2Processor, processor, source);
		} else {
			// Joint processor
			// to be implement.
		}
	}

	public void deleteOptimizedProcessor(HashMap<String, AbstractProcessor> id2Processor, AbstractProcessor processor,
			String source) throws CPSException {
		if (processor.getProcessorId().startsWith(source)) {
			// remove all sub successors in this optimized processor
			Set<String> subProcessors = ((OptimizedProcessor) processor).getSubprocessors();
			for (String sub : subProcessors) {
				id2Processor.remove(sub);
			}
			id2Processor.remove(processor.getProcessorId());
			processor.dispose();
			logger.info("Optimized Processor " + processor.getProcessorId() + " has been deleted.");
		} else {
			((OptimizedProcessor) processor).deleteSubProcessor(source);
			id2Processor.remove(source);
			logger.info("Subprocessor " + source + " of " + processor.getProcessorId() + " has been deleted.");
		}
	}

	public static boolean isProcessorType(String type) {
		return type2Constructor.keySet().contains(type);
	}

	public static String getIdFieldName(String type) {
		return MessageFields.PROCESSORID;
	}
	
	public static void main(String[] args) {
		System.out.println(type2Constructor.keySet());
	}
	
	public static boolean isInput(String type){
		return type.equals(MessageFields.SIMPLE_INPUT_TYPE) || type.equals(MessageFields.INPUT_TYPE);
	}
	
	public static boolean isStaticData(String type){
		return type.equals(MessageFields.STATIC_JOINT);
	}
	
}
